package com.lefu.risk.storm.bolt;


import com.lefu.risk.storm.common.constants.Constant;
import com.lefu.risk.storm.utils.JedisPoolUtil;
import org.apache.storm.shade.org.apache.commons.lang.StringUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by mwang on 2016/12/29.
 */
public class ResultReportBolt extends BaseRichBolt {

    private Map<String,Long> errorCodeMap = null;
    private JedisPoolUtil jedisPoolUtil = null;
    private OutputCollector collector;



    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.errorCodeMap = new HashMap<String,Long>();
        jedisPoolUtil = new JedisPoolUtil(Constant.REDIS_PROPERTIES_NAME);
    }

    @Override
    public void execute(Tuple input) {
        try {
            long resultDataBoltTime = input.getLongByField("resultDataBoltTime");
            String errorCode = input.getStringByField("errorCode");
            Long errorCodeCount = input.getLongByField("errorCodeCount");
            this.errorCodeMap.put(errorCode, errorCodeCount);
            errorCodeMap.forEach((k,v) -> {
                jedisPoolUtil.hset(Constant.ERROR_CODES_COUNT_REDIS_KEY,k,v.toString());
            });

            jedisPoolUtil.hset(Constant.RESULT_DATA_TIME_CONSUMING_REDIS_KEY,input.getStringByField("id"),""+((Long.parseLong(jedisPoolUtil.time().get(0)) * Constant.SECOND_TO_MICROSECONDS + Long.parseLong(jedisPoolUtil.time().get(1))) - resultDataBoltTime)/1000);
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {}

    @Override
    public void cleanup() {
    	System.out.println(errorCodeMap);
    }
}
